package com.spring.example2.listener;

import com.alibaba.fastjson.JSON;
import com.alibaba.fastjson.TypeReference;
import com.spring.example2.MqMessage;
import lombok.extern.slf4j.Slf4j;
import org.apache.rocketmq.common.message.MessageExt;
import org.apache.rocketmq.spring.core.RocketMQListener;

import java.nio.charset.StandardCharsets;

/**
 * author: chics
 * date: 2025/11/10
 */
@Slf4j
public abstract class BaseRocketmqListener<T> implements RocketMQListener<MessageExt> {

    @Override
    public void onMessage(MessageExt message) {
        long startTime = System.currentTimeMillis();
        String messageId = message.getMsgId();
        String topic = message.getTopic();
        String tag = message.getTags();
        String consumerGroup = message.getProperty("CONSUMER_GROUP");
        MqMessage<T> body = null;
        try {
            log.info("开始处理消息, messageId: {}, topic: {}, group:{}, tag: {}", messageId, topic,consumerGroup, tag);
            body = covert(message);
            // 前置处理
            if (!preProcess(body,message)){
                return;
            }
            // 业务处理
            if (handleMessage(body,message)) {
                log.info("消息处理成功, messageId: {}, topic: {}, tag: {}, cost: {}ms", messageId, topic, tag, System.currentTimeMillis() - startTime);
            } else {
                log.warn("消息处理失败, messageId: {}, topic: {}, tag: {}", messageId, topic, tag);
                // 可在此处添加重试逻辑或发送到死信队列
                if (handleConsumeFailure(body,message, "业务处理返回失败")){
                    retry(body, message, "业务处理返回失败");
                }
            }
        } catch (Exception e) {
            log.error("消息处理异常, messageId: {}, topic: {}, tag: {}, error: {}", messageId, topic, tag, e);
            if (handleConsumeException(body,message, e)){
                retry(body, message, e.getMessage());
            }
        }
    }

    /**
     * 消息转换（子类可重写）
     */
    protected MqMessage<T> covert(MessageExt message) {
        //return JSON.parseObject(new String(message.getBody(), StandardCharsets.UTF_8), getType());
        return JSON.parseObject(new String(message.getBody(), StandardCharsets.UTF_8), new TypeReference<MqMessage<T>>(){}.getType());
    }

    /**
     * 前置处理（子类可重写）
     * @return true: 继续处理，false: 停止处理
     */
    protected boolean preProcess(MqMessage<T> body, MessageExt message) {
        // 消息去重校验
        if (isMessageProcessed(body,message)) {
            log.info("消息已处理过，直接跳过, messageId: {}, topic: {}", message.getMsgId(), message.getTopic());
            return false;
        }
        return true;
    }

    /**
     * 检查消息是否已处理（幂等性检查）(可重写)
     * @return true: 已处理，false: 未处理
     */
    protected boolean isMessageProcessed(MqMessage<T> body, MessageExt message) {
        // 实现思路：
        // 1. 使用Redis记录已处理的消息ID，设置过期时间
        // 2. 使用数据库唯一索引
        // 3. 根据业务场景实现
        return false; // 默认返回false，子类根据实际情况实现
    }

    /**
     * 处理消息消费失败(可重写)
     * @return true: 继续重试，false: 停止重试
     */
    protected boolean handleConsumeFailure(MqMessage<T> body, MessageExt message, String errorMsg) {
        // 记录消费失败日志
        log.error("消息消费失败, messageId: {}, topic: {}, error: {}", message.getMsgId(), message.getTopic(), errorMsg);
        //retry(message, errorMsg);
        // 可在此处实现：
        // 1. 记录到数据库
        // 2. 发送告警
        // 3. 重试机制
        // 4. 发送到死信队列
        return true;
    }

    /**
     * 处理消息消费异常(可重写)
     * @return true: 继续重试，false: 停止重试
     */
    protected boolean handleConsumeException(MqMessage<T> body, MessageExt message, Exception e){
        // 记录消费异常日志
        log.error("消息消费异常, messageId: {}, topic: {}, error: {}", message.getMsgId(), message.getTopic(), e.getMessage());
        //retry(message, e.getMessage());
        // 可在此处实现：
        // 1. 记录到数据库
        // 2. 发送告警
        // 3. 重试机制
        // 4. 发送到死信队列
        return true;
    }

    /**
     * 重试(rocketmq抛出异常会自动重试)(可重写)
     */
    protected void retry(MqMessage<T> body, MessageExt message, String errorMsg) {
        if (message.getReconsumeTimes() >= getMaxRetryCount()) {
            stopRetry(body, message, errorMsg);
            log.error("消息重试次数已超过最大限制[{}]，不再进行重试, messageId: {}, topic: {}, tag: {}",message.getReconsumeTimes(), message.getMsgId(), message.getTopic(), message.getTags());
            return;
        }
        log.info("消息开始进行重试[{}], messageId: {}, topic: {}, tag: {}",message.getReconsumeTimes(), message.getMsgId(), message.getTopic(), message.getTags());
        throw new RuntimeException(errorMsg);
    }

    /**
     * 停止重试(可重写)
     */
    protected void stopRetry(MqMessage<T> body, MessageExt message, String errorMsg) {
        log.error("消息停止进行重试, messageId: {}, topic: {}, tag: {}", message.getMsgId(), message.getTopic(), message.getTags());
    }

    /**
     * 最大重试次数(可重写)，默认3次
     */
    protected int getMaxRetryCount() {
        return 3;
    }

    /**
     * 具体的消息处理方法（子类必须实现）
     * @return true: 处理成功，false: 处理失败
     */
    protected abstract boolean handleMessage(MqMessage<T> body, MessageExt messageExt);

    /**
     * 获取数据类型（子类必须实现）
     * 方法返回：复杂类型：new TypeReference<MqMessage<Map<String, Object>>>(){}.getType();
     *  简单类型：String.class
     *  数组类型：new TypeReference<MqMessage<List<String>>>(){}.getType();
     *  泛型类型：new TypeReference<MqMessage<TestPojo>>(){}.getType();
     *  嵌套泛型类型：new TypeReference<MqMessage<List<TestPojo>>>(){}.getType();
     */
    //protected abstract Type getType();
}
